package com.tmt.im.service.config;

import com.tmt.im.common.bean.DataSourceConfig;
import com.tmt.im.common.pojo.*;
import com.tmt.im.common.service.YunXinEventService;
import com.tmt.im.common.store.*;
import com.tmt.im.service.helper.MySqlHelper;
import com.tmt.im.service.interfaces.IDataSender;
import com.tmt.im.service.interfaces.impl.ChatDataSender;
import com.tmt.im.service.websocket.SessionContextManager;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.apache.ignite.*;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.QueryEntity;
import org.apache.ignite.cache.query.FieldsQueryCursor;
import org.apache.ignite.cache.query.SqlFieldsQuery;
import org.apache.ignite.cache.store.CacheStoreAdapter;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.lang.IgniteBiPredicate;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.StringUtils;

import javax.cache.configuration.FactoryBuilder;
import java.sql.*;
import java.text.MessageFormat;
import java.time.OffsetDateTime;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * Ignite配置
 *
 * @Description TODO
 * @Author wuyi
 * @Date 2024/7/16 15:18
 * @Version 1.0
 **/

@Slf4j
@Configuration
public class IgniteConfig {

    private final DataSourceProperties dataSourceProperties;

    @Value("${ignite.Chat-data.expire-days:3}")
    private int expireDays;

    @Value("${ignite.Chat-data.push-count:20}")
    private int pushCount;

    private IgniteCache<SessionMemberKey, ChatSessionMemberPoJo> sessionMemberCache;
    private IgniteCache<ChatDataKey, ChatDataPoJo> dataCache;

    private IgniteCache<Long, MemberStatusPoJo> memberStatusCache;

    private final SessionContextManager contextManager;

    private final ChatDataSender defaultDataSender;

    private final IgniteBiPredicate<UUID, ChatDataKey> defaultDataProcessor;

    private ConcurrentHashMap<Long, UUID> dataListenerMap;

    private IgniteMessaging igniteMessaging;

    private final String igniteConfigFile;

    public IgniteConfig(@Value("${ignite.config-xml:classpath:config/ignite-config.xml}")
                        String igniteConfigFile,
                        DataSourceProperties dataSourceProperties,
                        SessionContextManager contextManager) {
        this.igniteConfigFile = igniteConfigFile;
        this.dataSourceProperties = dataSourceProperties;
        this.contextManager = contextManager;
        this.defaultDataSender = new ChatDataSender();
        this.defaultDataProcessor = initDataProcessor();
    }

    @PreDestroy
    public void unregisterDataListener() {
        log.info("---停止ignite中监听云信抄送数据的消息队列---");
        for (UUID listener : dataListenerMap.values()) {
            igniteMessaging.stopRemoteListen(listener);
        }
    }

    @Bean
    public Ignite igniteInstance(@Qualifier("dataListenerMap") ConcurrentHashMap<Long, UUID> dataListenerMap) {
        Ignite ignite = Ignition.start(igniteConfigFile);

        //使用全局信号量确保以下初始化过程仅被执行一次。
        try (IgniteSemaphore semaphore = ignite.semaphore("systemSyn", 1, true, true)) {
            if (semaphore.tryAcquire(3, TimeUnit.SECONDS)) {
                try {
                    IgniteCache<String, Object> configParamCache = ignite.cache(Constant.APPLICATION_CONFIG_NAME);
                    if (configParamCache == null) {
                        //创建全局参数缓存。
                        configParamCache = ignite.getOrCreateCache(Constant.APPLICATION_CONFIG_NAME);
                    }

                    Object obj = configParamCache.get(Constant.DATA_SOURCE_CONFIG_NAME);
                    if (obj == null) {
                        DataSourceConfig config = new DataSourceConfig(dataSourceProperties.getUrl(),
                                dataSourceProperties.getUsername(), dataSourceProperties.getPassword());
                        //将数据源配置放入全局参数缓存中。
                        configParamCache.put(Constant.DATA_SOURCE_CONFIG_NAME, config);
                    }

                    memberStatusCache = ignite.cache(MemberStatusPoJo.CACHE_NAME);
                    if (memberStatusCache == null) {
                        //创建聊天用户状态缓存
                        memberStatusCache = ignite.getOrCreateCache(MemberStatusPoJo.CACHE_NAME);
                    }

                    //监听ignite转发的云信平台的抄送数据。
                    igniteMessaging = ignite.message(ignite.cluster().forRemotes());
                    UUID uuid = igniteMessaging.remoteListen(com.tmt.im.common.Constant.YUN_XIN_MSG_TOPIC_ID, new YunXinEventService());
                    dataListenerMap.put(com.tmt.im.common.Constant.YUN_XIN_MSG_TOPIC_ID, uuid);

                    IgniteCache<Long, IMUserPoJo> cache = ignite.cache(IMUserPoJo.CACHE_NAME);
                    if (cache == null) {
                        //初始化用户对象缓存,并将现有会话记录加载到缓存中。
                        initialCache(ignite, IMUserPoJo.CACHE_NAME, CacheAtomicityMode.TRANSACTIONAL, Long.class,
                                IMUserPoJo.class, IMUserStore.class, null, null);
                    }

                    IgniteCache<Long, ChatSessionPoJo> cache1 = ignite.cache(ChatSessionPoJo.CACHE_NAME);
                    if (cache1 == null) {
                        //初始化聊天会话对象缓存,并将现有会话记录加载到缓存中。
                        String initLoadSQL = MessageFormat.format("select * from {0}", ChatSessionPoJo.CACHE_NAME);
                        initialCache(ignite, ChatSessionPoJo.CACHE_NAME, CacheAtomicityMode.TRANSACTIONAL, Long.class,
                                ChatSessionPoJo.class, ChatSessionStore.class, initLoadSQL, null);
                    }

                    IgniteCache<Long, ChatSessionMemberPoJo> cache2 = ignite.cache(ChatSessionMemberPoJo.CACHE_NAME);
                    if (cache2 == null) {
                        //初始化聊天成员对象缓存,并将现有会话成员记录加载到缓存中。
                        String initLoadSQL = MessageFormat.format("select * from {0}", ChatSessionMemberPoJo.CACHE_NAME);
                        sessionMemberCache = initialCache(ignite, ChatSessionMemberPoJo.CACHE_NAME, CacheAtomicityMode.TRANSACTIONAL, SessionMemberKey.class,
                                ChatSessionMemberPoJo.class, ChatSessionMemberStore.class, initLoadSQL, null);
                    }

                    IgniteCache<Long, ChatDataPoJo> cache3 = ignite.cache(ChatDataPoJo.CACHE_NAME);
                    if (cache3 == null) {
                        //初始化聊天数据对象缓存,并将现有未过期的聊天记录加载到缓存中。
                        long validTime = Timestamp.from(OffsetDateTime.now().minusDays(expireDays).toInstant()).getTime();
                        String initLoadSQL = MessageFormat.format("select * from {0} where create_time > ?",
                                ChatDataPoJo.CACHE_NAME);
                        dataCache = initialCache(ignite, ChatDataPoJo.CACHE_NAME, CacheAtomicityMode.TRANSACTIONAL, ChatDataKey.class,
                                ChatDataPoJo.class, ChatDataStore.class, initLoadSQL, new Object[]{validTime});
                    }

                    IgniteCache<RedPackageDetailKey, RedPackageDetailPoJo> cache4 = ignite.cache(RedPackageDetailPoJo.CACHE_NAME);
                    if (cache4 == null) {
                        //初始化红包领取详情记录缓存数据
                        String initLoadSQL = MessageFormat.format("select * from {0}", RedPackageDetailPoJo.CACHE_NAME);
                        initialCache(ignite, RedPackageDetailPoJo.CACHE_NAME, CacheAtomicityMode.TRANSACTIONAL, RedPackageDetailKey.class,
                                RedPackageDetailPoJo.class, RedPackageDetailStore.class, initLoadSQL, null);
                    }

                    IgniteCache<Long, YunXinEventPoJo> cache5 = ignite.cache(YunXinEventPoJo.CACHE_NAME);
                    if (cache5 == null) {
                        initialCache(ignite, YunXinEventPoJo.CACHE_NAME, CacheAtomicityMode.ATOMIC, Long.class,
                                YunXinEventPoJo.class, YunXinEventStore.class, null, null);
                    }

                    IgniteCache<BlackBillKey, BlackBillPoJo> cache6 = ignite.cache(BlackBillPoJo.CACHE_NAME);
                    if (cache6 == null) {
                        initialCache(ignite, BlackBillPoJo.CACHE_NAME, CacheAtomicityMode.ATOMIC, BlackBillKey.class,
                                BlackBillPoJo.class, BlackBillStore.class, null, null);
                    }
                } catch (Exception ex) {
                    log.error("----Ignite数据初始化失败----" + ex.getMessage());
                }
            } else {
                log.error("获取全局信号量失败");
            }
        }

        this.contextManager.setIgnite(ignite);
        this.defaultDataSender.init(sessionMemberCache, memberStatusCache, dataCache, contextManager, pushCount);

        return ignite;
    }

    @Bean("dataListenerMap")
    public ConcurrentHashMap<Long, UUID> dataListenerMap() {
        dataListenerMap = new ConcurrentHashMap<>();
        return dataListenerMap;
    }

    @Bean("chatDataSender")
    public IDataSender chartDataSender() {
        return defaultDataSender;
    }

    /**
     * Ignite消息监听器的消息处理器方法。
     *
     * @return
     */
    @Bean("dataProcessor")
    public IgniteBiPredicate<UUID, ChatDataKey> dataProcessor() {
        return defaultDataProcessor;
    }

    private IgniteBiPredicate<UUID, ChatDataKey> initDataProcessor() {
        return (nodeId, msg) -> {
            List<ChatSessionMemberPoJo> sessionMemberPoJos = new ArrayList<>();
            try {
                //查询聊天群组中所有的用户ID。
                String sql = MessageFormat.format("select * from {0} where sessionId = ? and agreeStatus = ?",
                        ChatSessionMemberPoJo.CACHE_NAME);
                SqlFieldsQuery query = new SqlFieldsQuery(sql).setArgs(msg.getSessionId(), 1);
                try (FieldsQueryCursor<List<?>> cursor = sessionMemberCache.query(query)) {
                    StoreSqlHelper.convertTo(cursor, sessionMemberPoJos, ChatSessionMemberPoJo.class);
                }
            } catch (Exception ex) {
                log.error("查询群组中用户信息出现异常", ex);
            }
            //并发向聊天中的用户推送最新聊天数据。
            sessionMemberPoJos.parallelStream().forEach(defaultDataSender::sendData);
            return !sessionMemberPoJos.isEmpty();
        };
    }

    private <K, T extends IGetKey<K>> IgniteCache<K, T> initialCache(Ignite ignite, String cacheName, CacheAtomicityMode mode,
                                                                     Class<K> keyClazz, Class<T> valueClazz,
                                                                     Class<? extends CacheStoreAdapter<K, T>> storeClass,
                                                                     String initLoadSQL, Object[] sqlParams) {
        CacheConfiguration<K, T> cacheCfg = new CacheConfiguration<>();
        cacheCfg.setName(cacheName);
        cacheCfg.setCacheMode(CacheMode.PARTITIONED);
        cacheCfg.setAtomicityMode(mode);
        cacheCfg.setReadThrough(true);
        cacheCfg.setWriteThrough(true);
        cacheCfg.setCacheStoreFactory(FactoryBuilder.factoryOf(storeClass));
        cacheCfg.setBackups(1);

        //获取字段定义列表.
        LinkedHashMap<String, String> fields = StoreSqlHelper.getFieldDefineList(valueClazz);
        QueryEntity queryEntity = new QueryEntity()
                .setTableName(cacheName)
                .setKeyFieldName("id").setKeyType(keyClazz.getName()).setValueType(valueClazz.getName())
                .setFields(fields);

        cacheCfg.setQueryEntities(Collections.singletonList(queryEntity));
        IgniteCache<K, T> cache = ignite.getOrCreateCache(cacheCfg);

        //加载所有数据。
        if (StringUtils.hasText(initLoadSQL)) {
            log.info("-------向{}缓存加载数据-----开始------", cacheName);
            loadAll(ignite, cacheName, valueClazz, initLoadSQL, sqlParams);
            log.info("-------向{}缓存加载数据-----结束------", cacheName);
        }
        return cache;
    }

    private <K, T extends IGetKey<K>> void loadAll(Ignite ignite, String cacheName, Class<T> clazz,
                                                   String initLoadSQL, Object[] sqlParams) {
        try (IgniteDataStreamer<K, T> streamer = ignite.dataStreamer(cacheName)) {
            try (Connection conn = DriverManager.getConnection(dataSourceProperties.getUrl(), dataSourceProperties.getUsername(),
                    dataSourceProperties.getPassword())) {
                try (PreparedStatement st = conn.prepareStatement(initLoadSQL)) {
                    //初始化SQL中的参数
                    if (sqlParams != null) {
                        MySqlHelper.initPreparedStatement(st, sqlParams);
                    }
                    ResultSet result = st.executeQuery();
                    while (result.next()) {
                        T obj = clazz.getDeclaredConstructor().newInstance();
                        StoreSqlHelper.initialObject(result, obj);
                        streamer.addData(obj.getKey(), obj);
                        if (obj instanceof ChatSessionPoJo sessionPoJo) {
                            //创建群组数据监听器。
                            IgniteMessaging igniteMessaging = ignite.message(ignite.cluster().forRemotes());
                            UUID uuid = igniteMessaging.remoteListen(sessionPoJo.getId(), defaultDataProcessor);
                            ConcurrentHashMap<Long, UUID> dataListenerMap = dataListenerMap();
                            dataListenerMap.put(sessionPoJo.getId(), uuid);
                        }
                    }
                }
            } catch (Exception ex) {
                log.error(ex.getMessage(), ex);
            }
            //将流里面的剩余数据压进ignite
            streamer.flush();
        }

    }

}